package com.shujia.MR;


import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.HFileOutputFormat2;
import org.apache.hadoop.hbase.mapreduce.KeyValueSortReducer;
import org.apache.hadoop.hbase.mapreduce.LoadIncrementalHFiles;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;

import java.io.IOException;

/**
 * 需要使用MR来完成数据的导入工作
 *
 *  1.由于需要将文件导入至HDFS中并且转换成HFile形式，所以需要使用Map
 *  2.由于在这过程中不需要对数据进行做聚合操作，所以不需要用户自定义Reduce
 *  3.在Map端对文本文件数据进行读取操作，需要将其包装成HBASE存储数据的格式
 *
 */
public class MyBulkLoading {

    public static void main(String[] args) throws Exception {

        /**
         * 设置Job
         */
        Configuration conf = new Configuration();
        conf.set("hbase.zookeeper.quorum", "node1,node2,master");

        Job job = Job.getInstance(conf);
        job.setJarByClass(MyBulkLoading.class);
        job.setJobName("MyBulkLoading");

        // 设置Map
        job.setMapperClass(BulkMapper.class);
        job.setMapOutputKeyClass(ImmutableBytesWritable.class);
        job.setMapOutputValueClass(KeyValue.class);

        // 设置Reduce,由于HFILE中的RowKey需要满足字典序，所以需要设置Reduce
        job.setReducerClass(KeyValueSortReducer.class);

        // 设置输入路径 HDFS,由于生成的HFILE需要被HBASE使用，所以不能再本地执行，需要在集群中执行
        TextInputFormat.addInputPath(job,new Path("/data/dianxin/DIANXIN.csv"));
        Path hfOfDir = new Path("/data/dianxinHfile/");
        HFileOutputFormat2.setOutputPath(job,hfOfDir);

        // 创建HBASE连接，用于对数据进行做导入工作
        Connection hbCon = ConnectionFactory.createConnection(conf);
        String tableName = "api:bulkload";

        Table table = hbCon.getTable(TableName.valueOf(tableName));
        Admin admin = hbCon.getAdmin();

        // 获取表对应的Region信息
        RegionLocator regionLocator = hbCon.getRegionLocator(TableName.valueOf(tableName));

        // 配置HFILE对应的信息
        HFileOutputFormat2.configureIncrementalLoad(
                job
                ,table
                ,regionLocator
        );

        job.waitForCompletion(true);


        // 创建加载HFILE对应的类
        LoadIncrementalHFiles loadIncrementalHFiles = new LoadIncrementalHFiles(conf);

        // doBulkLoad函数需要在job提交之后去执行
        loadIncrementalHFiles.doBulkLoad(
                hfOfDir,admin,table,regionLocator
        );

    }



    /**
     * 由于MR读取的数据是本地文件系统中的文本文件
     *   对于Mapper的泛型
     *      由于读取本地文件系统那么 KeyIn ValueIn 为LongWritable, Text
     *      由于Map端直接将数据导出成HBASE所能识别的格式，那么对应数据类型为
     *          ImmutableBytesWritable: 对应存储KeyValue数据
     *          KeyValue:具体每个Cell对应的值
     *
     *  存储位置：
     *      api:bulkload  info1
     */
    public static class BulkMapper extends Mapper<LongWritable, Text, ImmutableBytesWritable, KeyValue>{
        /**
         *
         * @param key
         * @param value 文件中对应一行数据，对应HBASE中的一个RowKey下的数据
         */
        @Override
        protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, ImmutableBytesWritable, KeyValue>.Context context) throws IOException, InterruptedException {
            /**
             * CREATE TABLE IF NOT EXISTS DIANXIN (
             *      mdn VARCHAR ,
             *      start_date VARCHAR ,
             *      end_date VARCHAR ,
             *      county VARCHAR,
             *      x DOUBLE ,
             *      y  DOUBLE,
             *      bsid VARCHAR,
             *      grid_id  VARCHAR,
             *      biz_type VARCHAR,
             *      event_type VARCHAR ,
             *      data_source VARCHAR ,
             *      CONSTRAINT PK PRIMARY KEY (mdn,start_date)
             * ) column_encoded_bytes=0;
             * 47BE1E866CFC071DB19D5E1C056BE28AE24C16E7,20180503211049,20180503210349,8320113,118.908,32.063,2F3096801C8B19F6DFB82438C9A5E837,118905032060040,3,3,ddr
             */
            String[] columns = value.toString().split(",");
            // 1.取手机号码mdn+start_date作为ROWKEY
            ImmutableBytesWritable rowKey = new ImmutableBytesWritable(Bytes.toBytes(columns[0] + "*" + columns[1]));

            // 将其他列数据保存成一个KeyValue
            /**
             * public KeyValue(final byte [] row, final byte [] family,
             *       final byte [] qualifier, final byte [] value)
             */
            KeyValue start_date = new KeyValue(
                    Bytes.toBytes(columns[0] + "*" + columns[1])
                    , Bytes.toBytes("info1")
                    , Bytes.toBytes("start_date")
                    , Bytes.toBytes(columns[1])
            );

            context.write(rowKey,start_date);


            KeyValue end_date = new KeyValue(
                    Bytes.toBytes(columns[0] + "*" + columns[1])
                    , Bytes.toBytes("info1")
                    , Bytes.toBytes("end_date")
                    , Bytes.toBytes(columns[2])
            );

            context.write(rowKey,end_date);

            KeyValue county = new KeyValue(
                    Bytes.toBytes(columns[0] + "*" + columns[1])
                    , Bytes.toBytes("info1")
                    , Bytes.toBytes("county")
                    , Bytes.toBytes(columns[3])
            );

            context.write(rowKey,county);


            KeyValue grid_id = new KeyValue(
                    Bytes.toBytes(columns[0] + "*" + columns[1])
                    , Bytes.toBytes("info1")
                    , Bytes.toBytes("grid_id")
                    , Bytes.toBytes(columns[7])
            );

            context.write(rowKey,grid_id);
        }
    }
}
